package cn.weiwei.WHadoop.Avro;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.*;
import org.apache.avro.util.Utf8;

import java.io.ByteArrayOutputStream;
import java.io.IOException;

/**
 * <pre>字符序列化与反序列化</pre>
 * @author Wang Weiwei <weiwei02@vip.qq.com / weiwei.wang@100credit.com>
 * @version 1.0
 * @sine 17-4-6
 */
public class SerializeString {
    /**
     * 使用通用ＡＰＩ序列化字符串
     * */
    public ByteArrayOutputStream serialize(String path) throws IOException {
        Schema schema = buildSchema(path);


        //创建通用Ａｖｒｏ记录
        GenericRecord genericRecord = new GenericData.Record(schema);
        genericRecord.put("left",new Utf8("LEFT"));
        genericRecord.put("num",1000);
        genericRecord.put("right",new Utf8("RIGHT"));

        //将记录序列化到输出流中
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        //datumWriter对象将数据翻译成Encoder对象可以理解的类型然后写入到输出流
        DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<GenericRecord>(schema);
        //由于ｅｎｃｏｒｄ无需重用　binaryEncoder()的第二个参数为ｎｕｌｌ
        Encoder encoder = EncoderFactory.get().binaryEncoder(byteArrayOutputStream,null);
        datumWriter.write(genericRecord, encoder);
        encoder.flush();
        byteArrayOutputStream.close();
        return byteArrayOutputStream;
    }

    /**
     * 加载要使用的模式
     * */
    private Schema buildSchema(String path) throws IOException {
        //加载要使用的模式
        Schema.Parser parser = new Schema.Parser();
        return parser.parse(getClass().getClassLoader().getResourceAsStream(path));
    }

    /**使用通用ＡＰＩ反序列化数据*/
    public void deserialize(String path) throws IOException {
        Schema schema = buildSchema(path);
        DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>(schema);
        Decoder decoder = DecoderFactory.get().binaryDecoder(serialize(path).toByteArray(),null);
        GenericRecord record = datumReader.read(null,decoder);
        System.out.println(record.toString());
    }

}
